package com.chinaoly.baotou.web.controller;

import com.alibaba.fastjson.JSON;
import com.chinaoly.baotou.common.Result;
import com.chinaoly.baotou.common.utils.DateUtil;
import com.chinaoly.baotou.model.MessageParams;
import com.chinaoly.baotou.model.SubscribeNotifications;
import com.chinaoly.baotou.model.es.vo.ImageCaptureRecordEsVo;
import com.chinaoly.baotou.service.SmsService;
import com.chinaoly.frm.common.utils.CommonUtil;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.*;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

/**
 * Created by Administrator on 2023/9/16.
 */
@Slf4j
@RestController
@Api(tags = "kafka操作")
@RequestMapping("kafka")
public class KafkaController {

    @Autowired
    private KafkaTemplate kafkaTemplate;



    @Autowired
    private SmsService smsService;

    @Value("${spring.kafka.kafka-test.topic:framework-topic}")
    private String testTopic;

    @PostMapping("kafkaSend")
    @ApiOperation(value = "推送数据到Kafka",notes = "测试往kafka推送数据")
    public String test1018(@RequestBody String data){
        // 发送结果对象
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(testTopic, data);

        // 同步阻塞式获取发送结果：
        //try {
        // future.get()回调方法会阻塞，他会一直尝试获取发送结果，如果kafka迟迟没有返回发送结果那么程序会阻塞到这里
        // 所以这种发送方式是同步的，想要异步发送消息就不要这样获取发送结果
        //log.info("发送结果："+future.get().toString());
        //} catch (InterruptedException e) {
        //    throw new RuntimeException(e);
        //} catch (ExecutionException e) {
        //    throw new RuntimeException(e);
        //}

        // 异步非阻塞式获取发送结果：
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                log.error("消息 {} 发送至kafka，topic：{} 失败，原因：{}",data,testTopic,throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, String> stringStringSendResult) {
                log.info("消息 {} 成功发送至kafka，topic：{} ,offset: {}",data,testTopic,stringStringSendResult.getRecordMetadata().offset());
            }
        });
        return "success";
    }

    @PostMapping("sendMsg")
    @ApiOperation(value = "发送短信",notes = "测试通过短信通道发送预警信息功能")
    public String sendMsg(@RequestBody MessageParams messageParams){
        smsService.send(messageParams);
        return "success";
    }
}
